package com.jcloud.codetest1.component;

import com.jcloud.codetest1.domain.RecordItem;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * @author
 * @date 2021-04-16 8:51
 */
@Component
public class RecordTask  implements InitializingBean {
    private static final Logger LOG = LoggerFactory.getLogger(RecordTask.class);

    /**
     *存放处理任务的容器
     *将所有的recordItemList中的元素 送去由RecordComsumer进行消费
     *将元素的status状态都从N变成Y
     *如果消费返回U，代表消费者异常未处理，则需要再次送去消费
     *确保消费的次数dealNum 为1 ,既已经消费的数据不会重复消费
     *尽量保证消费的顺序的情况下，提高消费的效率,保证公平
     */
    public static List<RecordItem> recordItemList =  Collections.synchronizedList(new ArrayList<RecordItem>());

    //每隔5秒触发一次
    private long interval = 1000L;

    //延迟5秒多少启动
    private long initialDelay = 5000L;

    private ScheduledExecutorService service;
    private ScheduledExecutorService calcService;

    private RecordTask() {
        LOG.info("生成初始1000个任务开始");
        for(int i = 0;i<1000;i++){
            RecordItem recordItem = new RecordItem();
            recordItemList.add(recordItem);
        }
        LOG.info("生成初始1000个任务结束");
    }

    private void startGenRecord() {
        LOG.info("启动生成任务线程");
        this.service = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
            private final AtomicInteger count = new AtomicInteger(1);
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setDaemon(true);
                thread.setName(" genRecordThread-" + this.count.getAndIncrement());
                return thread;
            }
        });
        this.service.scheduleAtFixedRate(() -> {
            for(int i = 0;i< 10 ;i++){
                RecordItem recordItem = new RecordItem();
                recordItemList.add(recordItem);
            }
//            if(recordItemList.size() >= 1500){
//                this.service.shutdown();
//                LOG.info("添加任务结束：任务总数{}" ,recordItemList.size() );
//            }
        }, this.initialDelay, this.interval, TimeUnit.MILLISECONDS);
    }


    private void calcGenRecord() {
        this.calcService = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
            private final AtomicInteger count = new AtomicInteger(1);
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setDaemon(true);
                thread.setName(" calcThread-" + this.count.getAndIncrement());
                return thread;
            }
        });
        this.calcService.scheduleAtFixedRate(() -> {
            calcTask();
        }, 5000, 2000, TimeUnit.MILLISECONDS);
    }



    /**
     * 功能描述：统计不同状态的任务的个数
     *
     * @param: []
     * @Return: void
     * @Author:  2021/4/16 19:41
     */
    public void calcTask(){
        try {
            Map<String, Long> itemMap = RecordTask.recordItemList.stream().collect(
                    Collectors.groupingBy(item -> item.getStatus(), Collectors.counting()));
            LOG.info("--------------------");
            itemMap.forEach((k, v) -> {
                LOG.info("状态{} 数量 {}", k , v);
            });
            LOG.info("--------------------");
        }catch (Exception e) {
            //e.printStackTrace();
        }
    }



    @Override
    public void afterPropertiesSet() throws Exception {
        new Thread(() -> startGenRecord()).start();
        new Thread(() -> calcGenRecord()).start();
    }
}
